package com.butterfly.hotkey.worker.window;

import cn.hutool.core.date.SystemClock;
import cn.hutool.core.lang.Assert;

import java.util.concurrent.atomic.LongAdder;

/**
 * 基于LongAdder的滑动窗口
 *
 * @author butterfly
 */
public class SlidingWindow {

    /**
     * 默认窗口数量
     */
    private final static int DEFAULT_PANE_COUNT = 5;

    /**
     * 窗口的数量
     */
    private final int paneCount;

    /**
     * 阈值
     */
    private final int threshold;

    /**
     * 窗口的间隔
     */
    protected long paneIntervalInMs;

    /**
     * longAdder
     */
    private LongAdder[] longAdders;

    /**
     * 该滑窗的起始创建时间，也就是第一个数据
     */
    private long beginTimestamp;

    /**
     * 最后一个数据的时间戳
     */
    private long lastAddTimestamp;

    /**
     * 构建
     *
     * @param paneCount    窗口数量
     * @param intervalInMs 需要检测的时间间隔
     * @param threshold    检测的阈值
     */
    public SlidingWindow(int paneCount, long intervalInMs, int threshold) {
        Assert.isTrue(intervalInMs % paneCount == 0, "时间间隔必须是窗口数的整数倍");
        Assert.isTrue(paneCount > 1, "窗口数必须是正数");
        Assert.isTrue(threshold > 1, "阈值必须是正数");

        this.paneCount = paneCount;
        this.threshold = threshold;
        this.paneIntervalInMs = intervalInMs / paneCount;

        // 重置窗口
        reset();
    }

    /**
     * 构建时间窗口，默认窗口数5
     *
     * @param intervalInMs 需要检测的时间间隔
     * @param threshold    检测的阈值
     */
    public SlidingWindow(long intervalInMs, int threshold) {
        this(DEFAULT_PANE_COUNT, intervalInMs, threshold);
    }

    /**
     * 初始化时间窗口
     */
    private void reset() {
        // 用空间换时间, 使用2倍的空间存储访问次数
        beginTimestamp = SystemClock.now();
        int realPaneCount = paneCount * 2;
        LongAdder[] longAdders = new LongAdder[realPaneCount];
        for (int i = 0; i < realPaneCount; i++) {
            longAdders[i] = new LongAdder();
        }
        this.longAdders = longAdders;
    }

    /**
     * 增加一次访问
     * 同步接口synchronized
     *
     * @return 是否超过阈值
     */
    public synchronized boolean increment(long count) {
        //当前自己所在的位置，是哪个小时间窗
        int curIndex = currentPaneIndex();

        //然后清空自己前面windowSize到2*windowSize之间的数据格的数据
        clearInvalidPanes(curIndex);

        // 在当前时间片里增加次数
        LongAdder longAdder = longAdders[curIndex];
        longAdder.add(count);
        int sum = longAdder.intValue();

        // 加上前面几个窗口的访问数量
        for (int i = 1; i < paneCount; i++) {
            sum += longAdders[(curIndex - i + paneCount) % paneCount].intValue();
        }

        lastAddTimestamp = SystemClock.now();

        return sum >= threshold;
    }

    /**
     * 计算当前所在的时间片的位置
     */
    private int currentPaneIndex() {
        //如果当前的key已经超出一整个时间片了，那么就直接初始化就行了，不用去计算了
        long now = SystemClock.now();
        if (now - lastAddTimestamp > paneIntervalInMs * paneCount) {
            reset();
        }

        int index = (int) (((now - beginTimestamp) / paneIntervalInMs) % paneCount);
        return Math.max(index, 0);
    }

    /**
     * 清除无效窗口的计数
     * <p>
     * 譬如1秒分4个窗口，那么数组共计8个窗口
     * 当前index为5时，就清空6、7、8、1。然后把2、3、4、5的加起来就是该窗口内的总和
     *
     * @param index index
     */
    private void clearInvalidPanes(int index) {
        for (int i = 1; i <= paneCount; i++) {
            int j = index + i;
            if (j >= paneCount * 2) {
                j -= paneCount * 2;
            }
            longAdders[j].reset();
        }
    }
}
